package com.tonyodev.fetch2rx;

import ohos.eventhandler.EventHandler;
import ohos.utils.Pair;

import com.tonyodev.fetch2.CompletedDownload;
import com.tonyodev.fetch2.Download;
import com.tonyodev.fetch2.Error;
import com.tonyodev.fetch2.FetchConfiguration;
import com.tonyodev.fetch2.FetchErrorUtils;
import com.tonyodev.fetch2.FetchGroup;
import com.tonyodev.fetch2.FetchListener;
import com.tonyodev.fetch2.NetworkType;
import com.tonyodev.fetch2.Request;
import com.tonyodev.fetch2.Status;
import com.tonyodev.fetch2.database.DownloadInfo;
import com.tonyodev.fetch2.database.FetchDatabaseManagerWrapper;
import com.tonyodev.fetch2.exception.FetchException;
import com.tonyodev.fetch2.fetch.FetchHandler;
import com.tonyodev.fetch2.fetch.FetchModulesBuilder;
import com.tonyodev.fetch2.fetch.ListenerCoordinator;
import com.tonyodev.fetch2.util.ActiveDownloadInfo;
import com.tonyodev.fetch2.util.Defaults;
import com.tonyodev.fetch2.util.FetchUtils;
import com.tonyodev.fetch2.util.TypeConverterExtensions;
import com.tonyodev.fetch2core.DownloadBlock;
import com.tonyodev.fetch2core.Downloader;
import com.tonyodev.fetch2core.Extras;
import com.tonyodev.fetch2core.FetchObserver;
import com.tonyodev.fetch2core.FileResource;
import com.tonyodev.fetch2core.HandlerWrapper;
import com.tonyodev.fetch2core.Logger;
import com.tonyodev.fetch2core.Reason;
import com.tonyodev.fetch2rx.rxohos.OhosSchedulers;
import com.tonyodev.fetch2rx.util.RxUtils;

import org.jetbrains.annotations.NotNull;
import org.reactivestreams.Publisher;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import io.reactivex.Flowable;
import io.reactivex.Scheduler;
import io.reactivex.functions.Function;

/**
 * RxFetch实现类
 *
 * @since 2021-05-20
 */
public class RxFetchImpl implements RxFetch {
    private Scheduler scheduler;
    private Scheduler uiScheduler;
    private final Object lock = new Object();
    private volatile boolean closed;
    private Set activeDownloadsSet;
    private String namespace;
    private FetchConfiguration fetchConfiguration;
    private HandlerWrapper handlerWrapper;
    private EventHandler uiHandler;
    private FetchHandler fetchHandler;
    private Logger logger;
    private ListenerCoordinator listenerCoordinator;
    private FetchDatabaseManagerWrapper fetchDatabaseManagerWrapper;
    private Runnable activeDownloadsRunnable = new Runnable() {
        @Override
        public void run() {
            if (!isClosed()) {
                final boolean hasActiveDownloadsAdded = RxFetchImpl.this.fetchHandler.hasActiveDownloads(true);
                final boolean hasActiveDownloads = RxFetchImpl.this.fetchHandler.hasActiveDownloads(false);
                RxFetchImpl.this.uiHandler.postTask(new Runnable() {
                    @Override
                    public void run() {
                        if (!RxFetchImpl.this.isClosed()) {
                            Iterator iterator = RxFetchImpl.this.activeDownloadsSet.iterator();
                            ActiveDownloadInfo activeDownloadInfo = null;
                            boolean hasActive = false;
                            while (iterator.hasNext()) {
                                activeDownloadInfo = (ActiveDownloadInfo) iterator.next();
                                hasActive = activeDownloadInfo.getIncludeAddedDownloads() ? hasActiveDownloadsAdded
                                    : hasActiveDownloads;
                                activeDownloadInfo.getFetchObserver().onChanged(hasActive, Reason.REPORTING);
                            }
                        }

                        if (!isClosed()) {
                            registerActiveDownloadsRunnable();
                        }
                    }
                });
            }
        }
    };

    /**
     * 构造函数
     *
     * @param namespace 命名
     * @param fetchConfiguration 配置
     * @param handlerWrapper handler包装类
     * @param uiHandler UI线程Handler
     * @param fetchHandler fetchHandler
     * @param logger 日志包装类
     * @param listenerCoordinator ListenerCoordinator
     * @param fetchDatabaseManagerWrapper 数据库管理器包装类
     */
    public RxFetchImpl(String namespace, FetchConfiguration fetchConfiguration,
                       HandlerWrapper handlerWrapper, EventHandler uiHandler,
                       FetchHandler fetchHandler, Logger logger, ListenerCoordinator listenerCoordinator,
                       FetchDatabaseManagerWrapper fetchDatabaseManagerWrapper) {
        this.namespace = namespace;
        this.fetchConfiguration = fetchConfiguration;
        this.handlerWrapper = handlerWrapper;
        this.uiHandler = uiHandler;
        this.fetchHandler = fetchHandler;
        this.logger = logger;
        this.listenerCoordinator = listenerCoordinator;
        this.fetchDatabaseManagerWrapper = fetchDatabaseManagerWrapper;
        this.scheduler = OhosSchedulers.from(handlerWrapper.getEventRunner());
        this.uiScheduler = OhosSchedulers.mainThread();
        this.activeDownloadsSet = new LinkedHashSet();
        this.handlerWrapper.post(new Runnable() {
            @Override
            public void run() {
                fetchHandler.init();
            }
        });
        this.registerActiveDownloadsRunnable();
    }

    public ListenerCoordinator getListenerCoordinator() {
        return listenerCoordinator;
    }

    public FetchHandler getFetchHandler() {
        return fetchHandler;
    }

    private void registerActiveDownloadsRunnable() {
        handlerWrapper.postDelayed(activeDownloadsRunnable, fetchConfiguration.getActiveDownloadsCheckInterval());
    }

    @Override
    public boolean isClosed() {
        synchronized (lock) {
            return closed;
        }
    }

    @Override
    public Convertible<Request> enqueue(Request request) {
        Flowable flowable = this.enqueue(Collections.singletonList(request))
            .asFlowable()
            .subscribeOn(scheduler)
            .flatMap(new Function<List<Pair<Request, Error>>, Flowable<?>>() {
                @Override
                public Flowable<?> apply(List<Pair<Request, Error>> pairs) throws Exception {
                    if (pairs.isEmpty()) {
                        throw new FetchException("enqueue_not_successful");
                    }
                    Pair enqueuedPair = pairs.get(0);
                    if (enqueuedPair.s != Error.NONE) {
                        throw new FetchException("enqueue_not_successful");
                    }
                    return Flowable.just(enqueuedPair.f);
                }
            })
            .observeOn(this.uiScheduler);
        return RxUtils.toConvertible(flowable);
    }

    @Override
    public Convertible<List<Pair<Request, Error>>> enqueue(List<Request> requests) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(requests)
                .subscribeOn(scheduler)
                .flatMap(new Function<List<Request>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(List<Request> requests) throws Exception {
                        throwExceptionIfClosed();
                        int distinctCount = (int) requests.stream().count();
                        if (distinctCount != requests.size()) {
                            throw new FetchException("request_list_not_distinct");
                        }
                        List<Pair<Download, Error>> downloadPairs = fetchHandler.enqueue(requests);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Pair<Download, Error> downloadPair : downloadPairs) {
                                    Download download = downloadPair.f;
                                    switch (download.getStatus()) {
                                        case ADDED:
                                            listenerCoordinator.getMainListener().onAdded(download);
                                            break;
                                        case QUEUED:
                                            DownloadInfo downloadCopy = TypeConverterExtensions.toDownloadInfo(
                                                fetchDatabaseManagerWrapper.getNewDownloadInfoInstance(),
                                                new DownloadInfo());
                                            downloadCopy.setStatus(Status.ADDED);
                                            listenerCoordinator.getMainListener().onAdded(downloadCopy);
                                            listenerCoordinator.getMainListener().onQueued(download, false);
                                            break;
                                        case COMPLETED:
                                            listenerCoordinator.getMainListener().onCompleted(download);
                                            break;
                                        default:
                                            break;
                                    }
                                }
                            }
                        });
                        List<Pair<Request, Error>> requestList = new ArrayList<>();
                        if (downloadPairs != null && downloadPairs.size() > 0) {
                            for (Pair<Download, Error> pair : downloadPairs) {
                                requestList.add(new Pair(pair.f.getRequest(), pair.s));
                            }
                        }
                        return Flowable.just(requestList);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> pause(List<Integer> ids) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(ids)
                .subscribeOn(scheduler)
                .flatMap(new Function<List<Integer>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(List<Integer> integers) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.pause(ids);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    logger.d("Paused download $it");
                                    listenerCoordinator.getMainListener().onPaused(it);
                                }
                            }
                        });
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);

            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<Download> pause(int id) {
        Flowable flowable = pause(Collections.singletonList(id))
            .asFlowable()
            .subscribeOn(scheduler)
            .flatMap(new Function<List<Download>, Publisher<?>>() {
                @Override
                public Publisher<?> apply(List<Download> downloads) throws Exception {
                    if (downloads.isEmpty()) {
                        throw new FetchException("request_does_not_exist");
                    }
                    return Flowable.just(downloads.get(0));
                }
            })
            .observeOn(uiScheduler);
        return RxUtils.toConvertible(flowable);
    }

    @Override
    public Convertible<List<Download>> pauseGroup(int id) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(id)
                .subscribeOn(scheduler)
                .flatMap(new Function<Integer, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(Integer integer) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.pausedGroup(integer);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    logger.d("Paused download $download");
                                    listenerCoordinator.getMainListener().onPaused(it);
                                }
                            }
                        });
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<Boolean> freeze() {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(new Object())
                .subscribeOn(scheduler)
                .flatMap(new Function<Object, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(Object it) throws Exception {
                        throwExceptionIfClosed();
                        fetchHandler.freeze();
                        return Flowable.just(true);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> resume(List<Integer> ids) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(ids)
                .subscribeOn(scheduler)
                .flatMap(new Function<List<Integer>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(List<Integer> integers) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.resume(ids);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    logger.d("Queued download $it");
                                    listenerCoordinator.getMainListener().onQueued(it, false);
                                    listenerCoordinator.getMainListener().onResumed(it);
                                }
                            }
                        });
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<Download> resume(int id) {
        Flowable flowable = resume(Collections.singletonList(id))
            .asFlowable()
            .subscribeOn(scheduler)
            .flatMap(new Function<List<Download>, Publisher<?>>() {
                @Override
                public Publisher<?> apply(List<Download> downloads) throws Exception {
                    if (downloads.isEmpty()) {
                        throw new FetchException("request_does_not_exist");
                    }
                    return Flowable.just(downloads.get(0));
                }
            }).observeOn(uiScheduler);
        return RxUtils.toConvertible(flowable);
    }

    @Override
    public Convertible<List<Download>> resumeGroup(int id) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(id)
                .subscribeOn(scheduler)
                .flatMap(new Function<Integer, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(Integer integer) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.resumeGroup(integer);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    listenerCoordinator.getMainListener().onQueued(it, false);
                                    logger.d("Resumed download $download");
                                    listenerCoordinator.getMainListener().onResumed(it);
                                }
                            }
                        });
                        return null;
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<Boolean> unfreeze() {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(new Object())
                .subscribeOn(scheduler)
                .flatMap(new Function<Object, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(Object o) throws Exception {
                        throwExceptionIfClosed();
                        fetchHandler.unfreeze();
                        return Flowable.just(true);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> remove(List<Integer> ids) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(ids)
                .subscribeOn(scheduler)
                .flatMap(new Function<List<Integer>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(List<Integer> integers) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.remove(integers);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    logger.d("Removed download $download");
                                    listenerCoordinator.getMainListener().onRemoved(it);
                                }
                            }
                        });
                        return null;
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<Download> remove(int id) {
        Flowable flowable = remove(Collections.singletonList(id))
            .asFlowable()
            .subscribeOn(scheduler)
            .flatMap(new Function<List<Download>, Publisher<?>>() {
                @Override
                public Publisher<?> apply(List<Download> downloads) throws Exception {
                    if (downloads.isEmpty()) {
                        throw new FetchException("request_does_not_exist");
                    }
                    return Flowable.just(downloads.get(0));
                }
            })
            .observeOn(uiScheduler);
        return RxUtils.toConvertible(flowable);
    }

    @Override
    public Convertible<List<Download>> removeGroup(int id) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(id)
                .subscribeOn(scheduler)
                .flatMap(new Function<Integer, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(Integer integer) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.removeGroup(integer);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    logger.d("Removed download $download");
                                    listenerCoordinator.getMainListener().onRemoved(it);
                                }
                            }
                        });
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> removeAll() {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(new Object())
                .subscribeOn(scheduler)
                .flatMap(new Function<Object, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Object o) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.removeAll();
                        return null;
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> removeAllWithStatus(Status status) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(status)
                .subscribeOn(scheduler)
                .flatMap(new Function<Status, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Status status) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.removeAllWithStatus(status);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    logger.d("Removed download $download");
                                    listenerCoordinator.getMainListener().onRemoved(it);
                                }
                            }
                        });
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> removeAllInGroupWithStatus(int id, List<Status> status) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(new Pair<>(id, status))
                .subscribeOn(scheduler)
                .flatMap(new Function<Pair<Integer, List<Status>>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Pair<Integer, List<Status>> integerListPair) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.removeAllInGroupWithStatus(integerListPair.f,
                            integerListPair.s);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    logger.d("Removed download $download");
                                    listenerCoordinator.getMainListener().onRemoved(it);
                                }
                            }
                        });
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> delete(List<Integer> ids) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(ids)
                .subscribeOn(scheduler)
                .flatMap(new Function<List<Integer>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull List<Integer> integers) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.delete(ids);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    logger.d("Deleted download $it");
                                    listenerCoordinator.getMainListener().onDeleted(it);
                                }
                            }
                        });
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<Download> delete(int id) {
        Flowable flowable = delete(Collections.singletonList(id))
            .asFlowable()
            .subscribeOn(scheduler)
            .flatMap(new Function<List<Download>, Publisher<?>>() {
                @Override
                public Publisher<?> apply(@NotNull List<Download> downloads) throws Exception {
                    if (downloads.isEmpty()) {
                        throw new FetchException("request_does_not_exist");
                    }
                    return Flowable.just(downloads.get(0));
                }
            })
            .observeOn(uiScheduler);
        return RxUtils.toConvertible(flowable);
    }

    @Override
    public Convertible<List<Download>> deleteGroup(int id) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(id)
                .subscribeOn(scheduler)
                .flatMap(new Function<Integer, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Integer id) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.deleteGroup(id);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    logger.d("Deleted download $it");
                                    listenerCoordinator.getMainListener().onDeleted(it);
                                }
                            }
                        });
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> deleteAll() {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(new Object())
                .subscribeOn(scheduler)
                .flatMap(new Function<Object, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Object o) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.deleteAll();
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    logger.d("Deleted download $it");
                                    listenerCoordinator.getMainListener().onDeleted(it);
                                }
                            }
                        });
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> deleteAllWithStatus(Status status) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(status)
                .subscribeOn(scheduler)
                .flatMap(new Function<Status, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Status status) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.deleteAllWithStatus(status);
                        for (Download it : downloads) {
                            logger.d("Deleted download $download");
                            listenerCoordinator.getMainListener().onDeleted(it);
                        }
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> deleteAllInGroupWithStatus(int id, List<Status> statuses) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(new Pair<>(id, statuses))
                .subscribeOn(scheduler)
                .flatMap(new Function<Pair<Integer, List<Status>>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Pair<Integer, List<Status>> integerListPair) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.deleteAllInGroupWithStatus(integerListPair.f,
                            integerListPair.s);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    logger.d("Deleted download $download");
                                    listenerCoordinator.getMainListener().onDeleted(it);
                                }
                            }
                        });
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> cancel(List<Integer> ids) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(ids)
                .subscribeOn(scheduler)
                .flatMap(new Function<List<Integer>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull List<Integer> ids) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.cancel(ids);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    logger.d("Cancelled download $it");
                                    listenerCoordinator.getMainListener().onCancelled(it);
                                }
                            }
                        });
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<Download> cancel(int id) {
        Flowable flowable = cancel(Collections.singletonList(id))
            .asFlowable()
            .subscribeOn(scheduler)
            .flatMap(new Function<List<Download>, Publisher<?>>() {
                @Override
                public Publisher<?> apply(@NotNull List<Download> downloads) throws Exception {
                    if (downloads.isEmpty()) {
                        throw new FetchException("request_does_not_exist");
                    }
                    return Flowable.just(downloads.get(0));
                }
            })
            .observeOn(uiScheduler);
        return RxUtils.toConvertible(flowable);
    }

    @Override
    public Convertible<List<Download>> cancelGroup(int id) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(id)
                .subscribeOn(scheduler)
                .flatMap(new Function<Integer, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Integer id) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.cancelGroup(id);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    logger.d("Cancelled download $it");
                                    listenerCoordinator.getMainListener().onCancelled(it);
                                }
                            }
                        });
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> cancelAll() {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(new Object())
                .subscribeOn(scheduler)
                .flatMap(new Function<Object, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Object o) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.cancelAll();
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    logger.d("Cancelled download $it");
                                    listenerCoordinator.getMainListener().onCancelled(it);
                                }
                            }
                        });
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> retry(List<Integer> ids) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(ids)
                .subscribeOn(scheduler)
                .flatMap(new Function<List<Integer>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull List<Integer> ids) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.retry(ids);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    logger.d("Queued $it for download");
                                    listenerCoordinator.getMainListener().onQueued(it, false);
                                }
                            }
                        });
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<Download> retry(int id) {
        Flowable flowable = retry(Collections.singletonList(id))
            .asFlowable()
            .subscribeOn(scheduler)
            .flatMap(new Function<List<Download>, Publisher<?>>() {
                @Override
                public Publisher<?> apply(@NotNull List<Download> downloads) throws Exception {
                    if (downloads.isEmpty()) {
                        throw new FetchException("request_does_not_exist");
                    }
                    return Flowable.just(downloads.get(0));
                }
            })
            .observeOn(uiScheduler);
        return RxUtils.toConvertible(flowable);
    }

    @Override
    public Convertible<Download> resetAutoRetryAttempts(int downloadId, Boolean retryDownload) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(downloadId)
                .subscribeOn(scheduler)
                .flatMap(new Function<Integer, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Integer integer) throws Exception {
                        throwExceptionIfClosed();
                        Download download = fetchHandler.resetAutoRetryAttempts(downloadId, retryDownload);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                if (download != null && download.getStatus() == Status.QUEUED) {
                                    logger.d("Queued $download for download");
                                    listenerCoordinator.getMainListener().onQueued(download, false);
                                }
                            }
                        });
                        return Flowable.just(download);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<Download> updateRequest(int requestId, Request updatedRequest, Boolean notifyListeners) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(new Pair<>(requestId, updatedRequest))
                .subscribeOn(scheduler)
                .flatMap(new Function<Pair<Integer, Request>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Pair<Integer, Request> it) throws Exception {
                        throwExceptionIfClosed();
                        Pair<Download, Boolean> downloadPair = fetchHandler.updateRequest(it.f, it.s);
                        Download download = downloadPair.f;
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                if (notifyListeners) {
                                    switch (download.getStatus()) {
                                        case COMPLETED:
                                            listenerCoordinator.getMainListener().onCompleted(download);
                                            break;
                                        case FAILED:
                                            listenerCoordinator.getMainListener().onError(download, download.getError(),
                                                null);
                                            break;
                                        case CANCELLED:
                                            listenerCoordinator.getMainListener().onCancelled(download);
                                            break;
                                        case DELETED:
                                            listenerCoordinator.getMainListener().onDeleted(download);
                                            break;
                                        case PAUSED:
                                            listenerCoordinator.getMainListener().onPaused(download);
                                            break;
                                        case QUEUED:
                                            if (!downloadPair.s) {
                                                DownloadInfo downloadCopy = TypeConverterExtensions.toDownloadInfo(
                                                    download.copy(), new DownloadInfo());
                                                downloadCopy.setStatus(Status.ADDED);
                                                listenerCoordinator.getMainListener().onAdded(downloadCopy);
                                            }
                                            break;
                                        case REMOVED:
                                            listenerCoordinator.getMainListener().onRemoved(download);
                                            break;
                                        case DOWNLOADING:
                                            break;
                                        case ADDED:
                                            listenerCoordinator.getMainListener().onAdded(download);
                                            break;
                                        case NONE:
                                        default:
                                            break;
                                    }
                                }
                            }
                        });
                        return Flowable.just(download);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<Download> renameCompletedDownloadFile(int id, String newFileName) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(new Pair<>(id, newFileName))
                .subscribeOn(scheduler)
                .flatMap(new Function<Pair<Integer, String>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Pair<Integer, String> it) throws Exception {
                        throwExceptionIfClosed();
                        Download download = fetchHandler.renameCompletedDownloadFile(it.f, it.s);
                        return Flowable.just(download);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<Download> replaceExtras(int id, Extras extras) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(new Pair<>(id, extras))
                .subscribeOn(scheduler)
                .flatMap(new Function<Pair<Integer, Extras>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Pair<Integer, Extras> it) throws Exception {
                        throwExceptionIfClosed();
                        Download download = fetchHandler.replaceExtras(it.f, it.s);
                        return Flowable.just(download);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> getDownloads() {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(new Object())
                .subscribeOn(scheduler)
                .flatMap(new Function<Object, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Object o) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.getDownloads();
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> getDownloads(List<Integer> idList) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(idList)
                .subscribeOn(scheduler)
                .flatMap(new Function<List<Integer>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull List<Integer> idList) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.getDownloads(idList);
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<Download> getDownload(int id) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(id)
                .subscribeOn(scheduler)
                .flatMap(new Function<Integer, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Integer id) throws Exception {
                        throwExceptionIfClosed();
                        Download download = fetchHandler.getDownload(id);
                        return Flowable.just(download);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> getDownloadsInGroup(int groupId) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(groupId)
                .subscribeOn(scheduler)
                .flatMap(new Function<Integer, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Integer groupId) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.getDownloadsInGroup(groupId);
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> getDownloadsWithStatus(Status status) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(status)
                .subscribeOn(scheduler)
                .flatMap(new Function<Status, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Status status) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.getDownloadsWithStatus(status);
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> getDownloadsInGroupWithStatus(int groupId, List<Status> status) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(status)
                .subscribeOn(scheduler)
                .flatMap(new Function<List<Status>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull List<Status> status) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.getDownloadsInGroupWithStatus(groupId, status);
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<Download>> getDownloadsByRequestIdentifier(Long identifier) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(identifier)
                .subscribeOn(scheduler)
                .flatMap(new Function<Long, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Long aLong) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.getDownloadsByRequestIdentifier(identifier);
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<FetchGroup> getFetchGroup(int group) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(group)
                .subscribeOn(scheduler)
                .flatMap(new Function<Integer, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Integer integer) throws Exception {
                        throwExceptionIfClosed();
                        FetchGroup fetchGroup = fetchHandler.getFetchGroup(group);
                        return Flowable.just(fetchGroup);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public RxFetch addListener(FetchListener listener) {
        return addListener(listener, Defaults.DEFAULT_ENABLE_LISTENER_NOTIFY_ON_ATTACHED);
    }

    @Override
    public RxFetch addListener(FetchListener listener, Boolean notify) {
        return addListener(listener, notify, Defaults.DEFAULT_ENABLE_LISTENER_AUTOSTART_ON_ATTACHED);
    }

    @Override
    public RxFetch addListener(FetchListener listener, Boolean notify, Boolean autoStart) {
        synchronized (lock) {
            throwExceptionIfClosed();
            handlerWrapper.post(new Runnable() {
                @Override
                public void run() {
                    fetchHandler.addListener(listener, notify, autoStart);
                }
            });
            return this;
        }
    }

    @Override
    public RxFetch removeListener(FetchListener listener) {
        synchronized (lock) {
            throwExceptionIfClosed();
            handlerWrapper.post(new Runnable() {
                @Override
                public void run() {
                    fetchHandler.removeListener(listener);
                }
            });
            return this;
        }
    }

    @Override
    public Convertible<Download> addCompletedDownload(CompletedDownload completedDownload, Boolean alertListeners) {
        Flowable flowable = addCompletedDownloads(Collections.singletonList(completedDownload), alertListeners)
            .asFlowable()
            .subscribeOn(scheduler)
            .flatMap(new Function<List<Download>, Publisher<?>>() {
                @Override
                public Publisher<?> apply(@NotNull List<Download> downloads) throws Exception {
                    if (downloads.isEmpty()) {
                        throw new FetchException("failed_to_add_completed_download");
                    }
                    return Flowable.just(downloads.get(0));
                }
            })
            .observeOn(uiScheduler);
        return RxUtils.toConvertible(flowable);
    }

    @Override
    public Convertible<List<Download>> addCompletedDownloads(List<CompletedDownload> completedDownloads,
                                                             Boolean alertListeners) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(completedDownloads)
                .subscribeOn(scheduler)
                .flatMap(new Function<List<CompletedDownload>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull List<CompletedDownload> completedDownloads) throws Exception {
                        throwExceptionIfClosed();
                        List<Download> downloads = fetchHandler.enqueueCompletedDownloads(completedDownloads);
                        uiHandler.postTask(new Runnable() {
                            @Override
                            public void run() {
                                for (Download it : downloads) {
                                    if (alertListeners) {
                                        listenerCoordinator.getMainListener().onCompleted(it);
                                    }
                                    logger.d("Added CompletedDownload $download");
                                }
                            }
                        });
                        return Flowable.just(downloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<DownloadBlock>> getDownloadBlocks(int downloadId) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(downloadId)
                .subscribeOn(scheduler)
                .flatMap(new Function<Integer, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Integer integer) throws Exception {
                        throwExceptionIfClosed();
                        List<DownloadBlock> downloadBlocksList = fetchHandler.getDownloadBlocks(downloadId);
                        return Flowable.just(downloadBlocksList);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public RxFetch enableLogging(Boolean enabled) {
        synchronized (lock) {
            throwExceptionIfClosed();
            handlerWrapper.post(new Runnable() {
                @Override
                public void run() {
                    fetchHandler.enableLogging(enabled);
                }
            });
            return this;
        }
    }

    @Override
    public RxFetch setGlobalNetworkType(NetworkType networkType) {
        synchronized (lock) {
            throwExceptionIfClosed();
            handlerWrapper.post(new Runnable() {
                @Override
                public void run() {
                    fetchHandler.setGlobalNetworkType(networkType);
                }
            });
            return this;
        }
    }

    @Override
    public RxFetch setDownloadConcurrentLimit(int downloadConcurrentLimit) {
        synchronized (lock) {
            throwExceptionIfClosed();
            if (downloadConcurrentLimit < 0) {
                throw new FetchException("Concurrent limit cannot be less than 0");
            }
            handlerWrapper.post(new Runnable() {
                @Override
                public void run() {
                    fetchHandler.setDownloadConcurrentLimit(downloadConcurrentLimit);
                }
            });
            return this;
        }
    }

    @Override
    public void close() {
        synchronized (lock) {
            if (closed) {
                return;
            }
            closed = true;
            logger.d("$namespace closing/shutting down");
            handlerWrapper.removeCallbacks(activeDownloadsRunnable);
            handlerWrapper.post(new Runnable() {
                @Override
                public void run() {
                    try {
                        fetchHandler.close();
                    } catch (Exception e) {
                        logger.e("exception occurred whiles shutting down RxFetch with namespace:$namespace", e);
                    }
                }
            });
        }
    }

    @Override
    public Set<FetchListener> getListenerSet() {
        synchronized (lock) {
            throwExceptionIfClosed();
            return fetchHandler.getListenerSet();
        }
    }

    @Override
    public Convertible<Long> getContentLengthForRequest(Request request, Boolean fromServer) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(new Pair<>(request, fromServer))
                .subscribeOn(OhosSchedulers.from(handlerWrapper.getWorkTaskLooper()))
                .flatMap(new Function<Pair<Request, Boolean>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Pair<Request, Boolean> it) throws Exception {
                        throwExceptionIfClosed();
                        Long contentLength = fetchHandler.getContentLengthForRequest(it.f, it.s);
                        return Flowable.just(contentLength);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<Pair<List<Pair<Request, Long>>, List<Pair<Request, Error>>>>
    getContentLengthForRequests(List<Request> requests, Boolean fromServer) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(new Pair<>(requests, fromServer))
                .subscribeOn(OhosSchedulers.from(handlerWrapper.getWorkTaskLooper()))
                .flatMap(new Function<Pair<List<Request>, Boolean>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Pair<List<Request>, Boolean> listBooleanPair) throws Exception {
                        throwExceptionIfClosed();
                        List results = new ArrayList<Pair<Request, Long>>();
                        List results2 = new ArrayList<Pair<Request, Error>>();
                        for (Request request : requests) {
                            try {
                                results.add(new Pair<>(request,
                                    fetchHandler.getContentLengthForRequest(request, fromServer)));
                            } catch (Exception exception) {
                                logger.e("RxFetch with namespace $namespace error", exception);
                                Error error = FetchErrorUtils.getErrorFromMessage(exception.getMessage());
                                error.setThrowable(exception);
                                results2.add(new Pair<>(request, error));
                            }
                        }
                        return Flowable.just(new Pair<>(results, results2));
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<Downloader.Response> getServerResponse(String url, Map<String, String> headers) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(new Pair<>(url, headers))
                .subscribeOn(OhosSchedulers.from(handlerWrapper.getWorkTaskLooper()))
                .flatMap(new Function<Pair<String, Map<String, String>>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Pair<String,
                        Map<String, String>> stringMapPair) throws Exception {
                        throwExceptionIfClosed();
                        Downloader.Response contentLength = fetchHandler.getServerResponse(url, headers);
                        return Flowable.just(contentLength);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public Convertible<List<FileResource>> getFetchFileServerCatalog(Request request) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(request)
                .subscribeOn(OhosSchedulers.from(handlerWrapper.getWorkTaskLooper()))
                .flatMap(new Function<Request, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Request request) throws Exception {
                        throwExceptionIfClosed();
                        List<FileResource> catalogList = fetchHandler.getFetchFileServerCatalog(request);
                        return Flowable.just(catalogList);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public void awaitFinishOrTimeout(long allowTimeInMilliseconds) {
        FetchUtils.awaitFinishOrTimeout(allowTimeInMilliseconds, fetchHandler);
    }

    @Override
    public void awaitFinish() {
        awaitFinishOrTimeout(-1);
    }

    @Override
    public RxFetch attachFetchObserversForDownload(int downloadId, FetchObserver<Download> fetchObservers) {
        synchronized (lock) {
            throwExceptionIfClosed();
            handlerWrapper.post(new Runnable() {
                @Override
                public void run() {
                    fetchHandler.addFetchObserversForDownload(downloadId, fetchObservers);
                }
            });
            return this;
        }
    }

    @Override
    public RxFetch removeFetchObserversForDownload(int downloadId, FetchObserver<Download> fetchObservers) {
        synchronized (lock) {
            throwExceptionIfClosed();
            handlerWrapper.post(new Runnable() {
                @Override
                public void run() {
                    fetchHandler.removeFetchObserversForDownload(downloadId, fetchObservers);
                }
            });
            return this;
        }
    }

    @Override
    public Convertible<Boolean> hasActiveDownloads(Boolean includeAddedDownloads) {
        synchronized (lock) {
            throwExceptionIfClosed();
            Flowable flowable = Flowable.just(includeAddedDownloads)
                .subscribeOn(scheduler)
                .flatMap(new Function<Boolean, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(@NotNull Boolean it) throws Exception {
                        throwExceptionIfClosed();
                        Boolean hasActiveDownloads = fetchHandler.hasActiveDownloads(it);
                        return Flowable.just(hasActiveDownloads);
                    }
                })
                .observeOn(uiScheduler);
            return RxUtils.toConvertible(flowable);
        }
    }

    @Override
    public RxFetch addActiveDownloadsObserver(Boolean includeAddedDownloads, FetchObserver<Boolean> fetchObserver) {
        synchronized (lock) {
            throwExceptionIfClosed();
            handlerWrapper.post(new Runnable() {
                @Override
                public void run() {
                    activeDownloadsSet.add(new ActiveDownloadInfo(fetchObserver, includeAddedDownloads));
                }
            });
            return this;
        }
    }

    @Override
    public RxFetch removeActiveDownloadsObserver(FetchObserver<Boolean> fetchObserver) {
        synchronized (lock) {
            throwExceptionIfClosed();
            handlerWrapper.post(new Runnable() {
                @Override
                public void run() {
                    Iterator iterator = activeDownloadsSet.iterator();
                    while (iterator.hasNext()) {
                        ActiveDownloadInfo activeDownloadInfo = (ActiveDownloadInfo) iterator.next();
                        if (activeDownloadInfo.getFetchObserver() == fetchObserver) {
                            iterator.remove();
                            logger.d("Removed ActiveDownload FetchObserver $fetchObserver");
                            break;
                        }
                    }
                }
            });
            return this;
        }
    }

    private void throwExceptionIfClosed() {
        if (closed) {
            throw new FetchException("This rxFetch instance has been closed. Create a new "
                + "instance using the builder.");
        }
    }

    /**
     * 获取RxFetchImpl实例
     *
     * @param modules 模块包装类
     * @return RxFetchImpl实例
     */
    public static final RxFetchImpl newInstance(FetchModulesBuilder.Modules modules) {
        return new RxFetchImpl(modules.getFetchConfiguration().getNamespace(),
            modules.getFetchConfiguration(), modules.getHandlerWrapper(),
            modules.getUiHandler(), modules.getFetchHandler(),
            modules.getFetchConfiguration().getLogger(),
            modules.getListenerCoordinator(), modules.getFetchDatabaseManagerWrapper());
    }
}
